package com.cch.bigdata.flink.cep

import org.apache.flink.streaming.api.scala.{DataStream, KeyedStream, StreamExecutionEnvironment}
import org.apache.flink.api.scala._
import org.apache.flink.cep.functions.PatternProcessFunction
import org.apache.flink.cep.scala.CEP
import org.apache.flink.cep.scala.pattern.Pattern
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.util.Collector

import java.util

object CepDemo {

  //常量多少分钟内
  val timeInterval = 10

  //单位时间内出现的次数
  val batchSize = 10

  def main(args: Array[String]): Unit = {

    //获取执行环境
    val environment: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    environment.setParallelism(1)

    //使用处理好的数据
    //1用户10秒钟内点击了12次
    //2用户10秒钟点击了15次
    //3用户10秒钟点击了3次
    val dataStream: DataStream[String] = environment.readTextFile("data/ad_click.log")

    val keydStream: KeyedStream[AdClick, (String, String)] = dataStream.map(x => {
      val adClickLine: Array[String] = x.split(",")
      AdClick(adClickLine(0), adClickLine(1), adClickLine(2), adClickLine(3).toLong,adClickLine(4))
    }).assignAscendingTimestamps(_.timestamp).
      keyBy(x => (x.uid, x.productId))

    //定义模式
    val clickPattern: Pattern[AdClick, AdClick] = Pattern.begin[AdClick]("start").where(_.tag.equals("click")).within(Time.seconds(timeInterval))

    //应用模式
    val patternStream = CEP.pattern(keydStream, clickPattern)


    //提取数据
    patternStream.process(new PatternProcessFunction[AdClick,BlackUser](){

      //标记数据次数
      private val timeMap = new util.HashMap[String,Long]()

      override def open(parameters: Configuration): Unit = {}

      //关闭时，清理缓存
      override def close(): Unit = {
        timeMap.clear()
      }

      //从模式中选取
      override def processMatch(map: util.Map[String, util.List[AdClick]], context: PatternProcessFunction.Context, collector: Collector[BlackUser]): Unit = {
        val click: AdClick = map.get("start").get(0)
        val uid: String = click.uid
        val pid: String = click.productId
        val key = uid+"_"+pid;

        if(timeMap.containsKey(key)){
          timeMap.put(key,timeMap.get(key)+1)
        }else{
          timeMap.put(key,1)
        }

        val count: Long = timeMap.get(key)
        if(count>=batchSize){
          collector.collect(BlackUser(uid,pid,count))
        }
      }
    }).print()


    environment.execute()
  }
}
